//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// insert_executor.cpp
//
// Identification: src/execution/insert_executor.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//

#include <memory>
#include <vector>

#include "common/config.h"
#include "common/rid.h"
#include "execution/executors/insert_executor.h"
#include "storage/index/index.h"
#include "storage/table/tuple.h"
#include "type/value.h"
#include "type/value_factory.h"

namespace bustub {

InsertExecutor::InsertExecutor(ExecutorContext *exec_ctx, const InsertPlanNode *plan,
                               std::unique_ptr<AbstractExecutor> &&child_executor)
    : AbstractExecutor(exec_ctx),
      plan_(plan),
      child_executor_(std::move(child_executor)),
      table_info_(exec_ctx->GetCatalog()->GetTable(plan->table_oid_)) {}

void InsertExecutor::Init() {
  child_executor_->Init();
  sum_ = 0;
}

auto InsertExecutor::Next([[maybe_unused]] Tuple *tuple, RID *rid) -> bool {
  if (ending_) {
    return false;
  }
  Tuple value_tuple{};
  std::vector<Value> values;

  auto table_id = plan_->table_oid_;
  auto table_name = table_info_->name_;

  auto all_index_info = GetExecutorContext()->GetCatalog()->GetTableIndexes(table_name);

  auto status = child_executor_->Next(&value_tuple, rid);
  while (status) {
    auto insert_rid = this->GetExecutorContext()->GetCatalog()->GetTable(table_id)->table_->InsertTuple(
        {INVALID_TXN_ID, INVALID_TXN_ID, false}, value_tuple);
    if (!insert_rid.has_value() || insert_rid->GetPageId() == INVALID_PAGE_ID) {
      values.push_back(bustub::ValueFactory::GetIntegerValue(sum_));
      Tuple output_tuple{values, &GetOutputSchema()};
      return false;
    }

    for (auto index_info : all_index_info) {
      index_info->index_->InsertEntry(
          value_tuple.KeyFromTuple(table_info_->schema_, index_info->key_schema_, index_info->index_->GetKeyAttrs()),
          insert_rid.value(), GetExecutorContext()->GetTransaction());
    }
    sum_++;
    status = child_executor_->Next(&value_tuple, rid);
  }
  values.push_back(bustub::ValueFactory::GetIntegerValue(sum_));
  Tuple output_tuple{values, &GetOutputSchema()};
  *tuple = output_tuple;
  ending_ = true;
  return true;
}

}  // namespace bustub
